{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "86587ec8",
   "metadata": {},
   "source": [
    "importlib 的强大之处：\n",
    "\n",
    "- 动态导入：可以在运行时根据字符串路径导入模块\n",
    "- 模块创建：可以从文件路径创建模块对象\n",
    "- 灵活加载：支持插件式架构，无需修改主代码即可添加新功能"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3b907fe",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " 模拟工具目录结构:\n",
      "tools/\n",
      "├── math_tools.py\n",
      "├── text_tools.py\n",
      "customers/\n",
      "├── customer_a/tools/\n",
      "└── customer_b/tools/\n",
      "\n",
      " 启动动态 Agent (客户: customer_a)\n",
      "==================================================\n",
      " 工具目录不存在: tools/\n",
      " 为客户 customer_a 加载专属工具...\n",
      " 工具目录不存在: customers\\customer_a\\tools\n",
      " 工具目录不存在: tools/\n",
      " 工具目录不存在: customers/customer_a/tools/\n",
      " 动态加载完成，共 0 个工具: []\n",
      "\n",
      " 最终结果: 未找到可用工具\n"
     ]
    }
   ],
   "source": [
    "import importlib.util\n",
    "import os\n",
    "from typing import TypedDict, Dict, Callable\n",
    "\n",
    "# 1. 工具装饰器\n",
    "def tool(name: str, description: str = \"\"):\n",
    "    def decorator(func):\n",
    "        func.is_tool = True\n",
    "        func.tool_name = name\n",
    "        func.description = description\n",
    "        return func\n",
    "    return decorator\n",
    "\n",
    "# 2. 核心：动态加载器\n",
    "def load_tools_from_directory(tools_dir: str) -> Dict[str, Callable]:\n",
    "    \"\"\"\n",
    "    动态工具加载器 - 运行时扫描目录\n",
    "    与静态加载的核心区别\n",
    "    \"\"\"\n",
    "    tools = {}\n",
    "    \n",
    "    # 检查目录是否存在\n",
    "    if not os.path.exists(tools_dir):\n",
    "        print(f\" 工具目录不存在: {tools_dir}\")\n",
    "        return tools\n",
    "    \n",
    "    print(f\" 扫描工具目录: {tools_dir}\")\n",
    "    \n",
    "    # 动态扫描目录中的所有 Python 文件\n",
    "    for filename in os.listdir(tools_dir):\n",
    "        if filename.endswith('.py') and not filename.startswith('__'):\n",
    "            file_path = os.path.join(tools_dir, filename)\n",
    "            module_name = filename[:-3]  # 移除 .py 后缀\n",
    "            \n",
    "            try:\n",
    "                # importlib 动态加载模块\n",
    "                spec = importlib.util.spec_from_file_location(module_name, file_path)\n",
    "                module = importlib.util.module_from_spec(spec)\n",
    "                spec.loader.exec_module(module)\n",
    "                \n",
    "                # 动态发现模块中的工具函数\n",
    "                for attr_name in dir(module):\n",
    "                    obj = getattr(module, attr_name)\n",
    "                    if hasattr(obj, 'is_tool'):\n",
    "                        tools[obj.tool_name] = obj\n",
    "                        print(f\" 发现工具: {obj.tool_name} ({obj.description})\")\n",
    "                        \n",
    "            except Exception as e:\n",
    "                print(f\" 加载失败 {filename}: {e}\")\n",
    "    \n",
    "    return tools\n",
    "\n",
    "# 3. 多客户支持的动态加载器\n",
    "def load_customer_tools(customer_id: str, base_dir: str = \"customers\") -> Dict[str, Callable]:\n",
    "    \"\"\"\n",
    "    为特定客户动态加载工具 - 真实业务场景\n",
    "    \"\"\"\n",
    "    customer_tools_dir = os.path.join(base_dir, customer_id, \"tools\")\n",
    "    print(f\" 为客户 {customer_id} 加载专属工具...\")\n",
    "    return load_tools_from_directory(customer_tools_dir)\n",
    "\n",
    "# 4. 配置驱动的动态加载器\n",
    "def load_tools_by_config(config: dict) -> Dict[str, Callable]:\n",
    "    \"\"\"\n",
    "    根据配置动态加载工具 - 支持 A/B 测试\n",
    "    \"\"\"\n",
    "    tools = {}\n",
    "    \n",
    "    for tool_config in config.get(\"enabled_tools\", []):\n",
    "        tools_dir = tool_config[\"directory\"]\n",
    "        enabled_tools = tool_config.get(\"tools\", [])\n",
    "        \n",
    "        # 加载目录中的所有工具\n",
    "        dir_tools = load_tools_from_directory(tools_dir)\n",
    "        \n",
    "        # 根据配置筛选工具\n",
    "        if enabled_tools:\n",
    "            for tool_name in enabled_tools:\n",
    "                if tool_name in dir_tools:\n",
    "                    tools[tool_name] = dir_tools[tool_name]\n",
    "        else:\n",
    "            tools.update(dir_tools)\n",
    "    \n",
    "    return tools\n",
    "\n",
    "# 5. LangGraph 状态（保持不变）\n",
    "class AgentState(TypedDict):\n",
    "    task: str\n",
    "    tools: dict\n",
    "    result: str\n",
    "    customer_id: str  # 新增客户ID\n",
    "\n",
    "# 6. 动态的 LangGraph 节点\n",
    "def dynamic_load_tools_node(state: AgentState) -> AgentState:\n",
    "    \"\"\"\n",
    "    动态工具加载节点 - 核心区别\n",
    "    \"\"\"\n",
    "    customer_id = state.get(\"customer_id\", \"default\")\n",
    "    \n",
    "    # 方式1: 从通用工具目录加载，运行时扫描，动态发现\n",
    "    general_tools = load_tools_from_directory(\"tools/\")\n",
    "    \n",
    "    # 方式2: 从客户专属目录加载\n",
    "    customer_tools = load_customer_tools(customer_id)\n",
    "    \n",
    "    # 方式3: 根据配置加载\n",
    "    config = {\n",
    "        \"enabled_tools\": [\n",
    "            {\"directory\": \"tools/\", \"tools\": [\"calculator\", \"text_processor\"]},\n",
    "            {\"directory\": f\"customers/{customer_id}/tools/\"}\n",
    "        ]\n",
    "    }\n",
    "    config_tools = load_tools_by_config(config)\n",
    "    \n",
    "    # 合并所有工具（客户工具优先级更高）\n",
    "    all_tools = {}\n",
    "    all_tools.update(general_tools)\n",
    "    all_tools.update(customer_tools)\n",
    "    all_tools.update(config_tools)\n",
    "    \n",
    "    state[\"tools\"] = all_tools\n",
    "    print(f\" 动态加载完成，共 {len(all_tools)} 个工具: {list(all_tools.keys())}\")\n",
    "    \n",
    "    return state\n",
    "\n",
    "# 7. 热重载功能\n",
    "def hot_reload_tools(state: AgentState) -> AgentState:\n",
    "    \"\"\"\n",
    "    热重载工具 - 无需重启即可更新工具\n",
    "    \"\"\"\n",
    "    print(\" 执行热重载...\")\n",
    "    return dynamic_load_tools_node(state)\n",
    "\n",
    "# 8. 演示运行\n",
    "def run_dynamic_agent(task: str, customer_id: str = \"default\"):\n",
    "    \"\"\"运行真正动态的 Agent\"\"\"\n",
    "    state = AgentState(\n",
    "        task=task, \n",
    "        tools={}, \n",
    "        result=\"\", \n",
    "        customer_id=customer_id\n",
    "    )\n",
    "    \n",
    "    print(f\" 启动动态 Agent (客户: {customer_id})\")\n",
    "    print(\"=\" * 50)\n",
    "    \n",
    "    #### 硬编码，编译时确定\n",
    "    ###state[\"tools\"] = {\"calculator\": add_numbers}\n",
    "\n",
    "\n",
    "    # 动态加载工具\n",
    "    state = dynamic_load_tools_node(state)\n",
    "    \n",
    "    # 模拟工具执行\n",
    "    if state[\"tools\"]:\n",
    "        tool_name = list(state[\"tools\"].keys())[0]\n",
    "        print(f\" 使用工具: {tool_name}\")\n",
    "        state[\"result\"] = f\"使用 {tool_name} 处理任务: {task}\"\n",
    "    else:\n",
    "        state[\"result\"] = \"未找到可用工具\"\n",
    "    \n",
    "    return state\n",
    "\n",
    "# 演示\n",
    "if __name__ == \"__main__\":\n",
    "    # 创建示例工具文件内容（实际使用时这些应该是独立文件）\n",
    "    print(\" 模拟工具目录结构:\")\n",
    "    print(\"tools/\")\n",
    "    print(\"├── math_tools.py\")\n",
    "    print(\"├── text_tools.py\") \n",
    "    print(\"customers/\")\n",
    "    print(\"├── customer_a/tools/\")\n",
    "    print(\"└── customer_b/tools/\")\n",
    "    print()\n",
    "    \n",
    "    # 运行演示\n",
    "    result = run_dynamic_agent(\"处理数据\", \"customer_a\")\n",
    "    print(f\"\\n 最终结果: {result['result']}\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18895a8b",
   "metadata": {},
   "source": [
    "## 异常处理机制"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fd7a6efb",
   "metadata": {},
   "source": [
    "1. 工具失败重试\n",
    "\n",
    "使用 tenacity 库实现智能重试机制，\n",
    "\n",
    "在网络波动或临时故障时自动恢复。\n",
    "\n",
    "```python\n",
    "from tenacity import retry, stop_after_attempt, wait_exponential\n",
    "\n",
    "@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))\n",
    "def call_with_retry(tool, inputs):\n",
    "    return tool.invoke(inputs)\n",
    "````\n",
    "\n",
    "配置最多重试3次，间隔时间指数增长（1秒、2秒、4秒），避免雪崩效应。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f984b91c",
   "metadata": {},
   "source": [
    "2. 超时中断\n",
    "\n",
    "防止某个工具长时间阻塞整个对话流程。\n",
    "\n",
    "```python\n",
    "try:\n",
    "    result = await asyncio.wait_for(tool.run(), timeout=10.0)\n",
    "except asyncio.TimeoutError:\n",
    "    raise ToolExecutionTimeout(\"工具执行超时\")\n",
    "```\n",
    "\n",
    "设置合理超时时间（如10秒），超时后抛出异常并提示用户。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e5d3a2d2",
   "metadata": {},
   "source": [
    "3. 结果缓存机制\n",
    "\n",
    "对于频繁调用且结果不变的查询类工具，启用缓存以提升性能。\n",
    "\n",
    "```python\n",
    "cached_result = redis.get(f\"tool:{hash(inputs)}\")\n",
    "if cached_result:\n",
    "    return json.loads(cached_result)\n",
    "\n",
    "result = tool.invoke(inputs)\n",
    "redis.setex(f\"tool:{hash(inputs)}\", 300, json.dumps(result))  # 缓存5分钟\n",
    "```\n",
    "\n",
    "适用于订单查询、商品信息获取等幂等性操作，显著降低后端压力。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "64f9aeed",
   "metadata": {},
   "source": [
    "4. 文件监听与动态重载\n",
    "\n",
    "使用 watchdog 库监听插件目录变化，检测到代码修改后自动重载。\n",
    "\n",
    "```python\n",
    "from watchdog.observers import Observer\n",
    "from watchdog.events import FileSystemEventHandler\n",
    "\n",
    "class PluginReloader(FileSystemEventHandler):\n",
    "    def on_modified(self, event):\n",
    "        if event.src_path.endswith(\".py\"):\n",
    "            plugin_name = extract_plugin_name(event.src_path)\n",
    "            if plugin_name in loaded_plugins:\n",
    "                importlib.reload(loaded_plugins[plugin_name])\n",
    "                print(f\"插件 {plugin_name} 已重新加载\")\n",
    "```\n",
    "\n",
    "启动时运行监听器：\n",
    "```python\n",
    "observer = Observer()\n",
    "observer.schedule(PluginReloader(), path=\"plugins\", recursive=True)\n",
    "observer.start()\n",
    "```"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "MLOps",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
